package com.navinfo.tripanalysis.offline.service.impl.huashen;

import com.navinfo.tripanalysis.common.pojo.Point0F39;
import com.navinfo.tripanalysis.common.pojo.RunningType;
import com.navinfo.tripanalysis.common.pojo.RunningTypeEnum;
import com.navinfo.tripanalysis.common.util.CommonUtils;
import com.navinfo.tripanalysis.offline.pojo.LoadDataParam;
import com.navinfo.tripanalysis.offline.service.LoadPoint0F39Service;
import com.navinfo.tripanalysis.offline.service.impl.AbstractLoadPoint0F39Service;
import com.navinfo.tripanalysis.offline.util.PointUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
import scala.collection.JavaConverters;
import scala.collection.mutable.WrappedArray;

import java.util.*;

/**
 * 一汽通过hive加载0F39故障数据
 */
@RunningType(value = RunningTypeEnum.HS, interfaceClass = LoadPoint0F39Service.class )
public class LoadPoint0F39ServiceHiveImpl extends AbstractLoadPoint0F39Service {

    /**
     * 覆盖默认的TID字段
     */
    private String TID_NAME = "terminalid";

    public LoadPoint0F39ServiceHiveImpl() {
        COL_NAMES.addAll(Arrays.asList(TID_NAME,"gpslocationdata","faultlistdata",
                "originallat","originallng"));
    }
    @Override
    protected Iterator<Point0F39> getPoint(Row row) throws CloneNotSupportedException {
        Point0F39 point = new Point0F39();
        List<Point0F39> point0F39s = new ArrayList<>();
        //tid,终端号
        point.setTid(String.valueOf(Optional.ofNullable(row.getAs(TID_NAME)).orElse("0")));
        //协议号,0F39
        point.setProtocol("0F39");
        //84原始经度
        point.setGpsTime(String.valueOf(Optional.ofNullable(row.getAs("originallat")).orElse("0")));
        //84原始纬度
        point.setGpsTime(String.valueOf(Optional.ofNullable(row.getAs("originallng")).orElse("0")));
        //gpslocationdata,基本信息集合
        GenericRowWithSchema gpslocationRow = (GenericRowWithSchema)Optional.ofNullable(row.getAs("gpslocationdata")).orElse(null);
        if(gpslocationRow != null){
            praseGpsLocaltionData(point,gpslocationRow);
        }
        //故障明细信息
        scala.collection.mutable.Seq<GenericRowWithSchema> faultInfoArr =
                (WrappedArray.ofRef<GenericRowWithSchema>)Optional.ofNullable(row.getAs("faultlistdata")).orElse(null);
        if(faultInfoArr != null){
            List<GenericRowWithSchema> faultInfoList = new ArrayList<>();
            faultInfoList.addAll(JavaConverters.mutableSeqAsJavaListConverter(faultInfoArr).asJava());
            if(faultInfoList!=null && CommonUtils.isCollectionNotEmpty(faultInfoList)){
                point0F39s = parseFaultInfo(point,faultInfoList);
            }
        }
        else {
            point0F39s.add(point);
        }
        return point0F39s.iterator();
    }

    @Override
    protected String getWhereSql(LoadDataParam param) {
        List<Long> tidList = param.getTidList();

        //分批处理
        String batchWhere = PointUtils.getBatchWhereSqlWihtoutHashtid(param);
        StringBuilder sb = new StringBuilder(StringUtils.isEmpty(batchWhere) ? "" : batchWhere+" and ")
                .append(TID_NAME +" is not null ")
                .append(CollectionUtils.isNotEmpty(tidList) ? String.format(" and %s in(%s) ", TID_NAME, CommonUtils.mkString(tidList, ",")) : "");

        return sb.toString();

    }


    @Override
    protected String getHDFSWhereSql(LoadDataParam param){
        List<Long> tidList = param.getTidList();
        StringBuilder sb = new StringBuilder(TID_NAME +"!='' and "+ TID_NAME +" is not null ")
                .append(CollectionUtils.isNotEmpty(tidList) ? String.format(" and %s in(%s) ", TID_NAME, CommonUtils.mkString(tidList, ",")) : "");
        return sb.toString();
    }

    protected void praseGpsLocaltionData (Point0F39 point,GenericRowWithSchema gpslocationRow){

        //gpstime
        point.setGpsTime(String.valueOf(Optional.ofNullable(gpslocationRow.getAs("gpsTime")).orElse("0")));
        //经度
        point.setLongitude(String.valueOf(Optional.ofNullable(gpslocationRow.getAs("longitude")).orElse("0")));
        //纬度
        point.setLatitude(String.valueOf(Optional.ofNullable(gpslocationRow.getAs("latitude")).orElse("0")));
        //高程
        point.setHeight(String.valueOf(Optional.ofNullable(gpslocationRow.getAs("height")).orElse("0")));
        //方向
        point.setDirection(String.valueOf(Optional.ofNullable(gpslocationRow.getAs("direction")).orElse("0")));
        //速度
        point.setSpeed(String.valueOf(Optional.ofNullable(gpslocationRow.getAs("speed")).orElse("0")));

    }


    protected List<Point0F39> parseFaultInfo (Point0F39 point,List<GenericRowWithSchema> faultinfoList) throws CloneNotSupportedException {

        List<Point0F39> list =  new ArrayList<>();
        //按层级展开list
        for(GenericRowWithSchema faultinfo : faultinfoList){
            Point0F39 point0 = (Point0F39) point.clone();
            //故障类型
            point0.setFaultType(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("faultType")).orElse("0"))));
            //车速
            point0.setSubSpeed(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("subSpeed")).orElse(0))));
            //油门开度百分比
            point0.setAccelerator(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("accelerator")).orElse(0))));
            //制动信号状态
            point0.setBrake(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("brakeSignal")).orElse(0))));
            //发动机转速
            point0.setRotation(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("rotation")).orElse(0))));
            //发动机涡轮增压压力
            point0.setEngineTurbochargedPressure(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("engineTurbochargedPressure")).orElse(0))));
            //进气压力
            point0.setEngineIntakePressure(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("engineIntakePressure")).orElse(0))));
            //发动机排气温度
            point0.setEngineExhaustTemperature(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("engineExhaustTemperature")).orElse(0))));
            //发动机水温
            point0.setEngineCoolantTemperature(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("engineCoolantTemperature")).orElse(0))));
            //油门变化率
            point0.setAccelerationRate(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("accelerationRate")).orElse(0))));
            //挡位
            point0.setGear(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("gear")).orElse(0))));
            //发动机输出扭矩
            point0.setEngineOutputTorque(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("engineOutputTorque")).orElse(0))));
            //载荷
            point0.setLoad(Long.parseLong(String.valueOf(Optional.ofNullable(faultinfo.getAs("load")).orElse(0))));
            //发动机负荷
            point0.setEngineLoad(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("engineLoad")).orElse(0))));
            //车辆加速度
            point0.setVehicleAcceleration(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("vehicleAcceleration")).orElse(0))));
            //车辆减速度
            point0.setVehicleSpeedReduction(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("vehicleSpeedReduction")).orElse(0))));
            //故障码列表
            point0.setFaultCodeCount(Integer.parseInt(String.valueOf(Optional.ofNullable(faultinfo.getAs("faultCodeCount")).orElse(0))));

            //遍历故障地址项
            scala.collection.mutable.Seq<GenericRowWithSchema> faultItemArr = (WrappedArray.ofRef<GenericRowWithSchema>)Optional.ofNullable(faultinfo.getAs("faultItem")).orElse(null);
            if(faultItemArr == null){
                list.add(point0);
            }else{
                List<GenericRowWithSchema> faultItemList = JavaConverters.mutableSeqAsJavaListConverter(faultItemArr).asJava();
                for(GenericRowWithSchema faultItem : faultItemList){
                    Point0F39 point1 = (Point0F39) point0.clone();
                    //源地址
                    point1.setFaultAddr(Integer.parseInt(String.valueOf(Optional.ofNullable(faultItem.getAs("faultAddr")).orElse(0))));
                    //SPN
                    point1.setSpn(Integer.parseInt(String.valueOf(Optional.ofNullable(faultItem.getAs("SPN")).orElse(0))));
                    //FMI
                    point1.setFmi(Integer.parseInt(String.valueOf(Optional.ofNullable(faultItem.getAs("FMI")).orElse(0))));
                    list.add(point1);
                }
            }

        }
        return list;
    }

}
